package com.ezcloud.engine.server;

import com.ezcloud.engine.CustTMessage;
import com.ezcloud.engine.CustTMessageDecoder;
import com.ezcloud.engine.CustTMessageEncoder;
import com.ezcloud.engine.CustTType;
import com.ezcloud.engine.rpc.server.RpcServerHandler;
import com.ezcloud.engine.util.RemotingUtil;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.epoll.EpollChannelOption;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultEventExecutorGroup;
import io.netty.util.concurrent.DefaultThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;

/**
 * @author : zuodp
 * @version : 1.10
 */
public class NettyServer {
    private static Logger logger = LoggerFactory.getLogger(NettyServer.class);

    private NettyServerOptions serverOptions;

    private EventLoopGroup bossGroup;
    private EventLoopGroup workerGroup;
    private DefaultEventExecutorGroup defaultEventExecutorGroup;
    private ChannelFuture[] bindFuture;
    private NettyConnectManager connectManager = new NettyConnectManager();

    public NettyServer() {
        this(NettyServerOptions.DEFAULT);
    }

    public NettyServer(NettyServerOptions serverOptions) {
        this.serverOptions = serverOptions;
    }

    public void serve(int port, Supplier<? extends SimpleChannelInboundHandler<CustTMessage>> supplier) {
        logger.info("Netty Server is starting");

        ServerBootstrap b = configServer(supplier.get());

        try {
            // start server
            bindFuture = new ChannelFuture[this.serverOptions.getWorkers()];

            for (int i = 0; i < this.serverOptions.getWorkers(); i++) {
                bindFuture[i] = b.bind(port).sync();
            }

            // register shutdown hook
            Runtime.getRuntime().addShutdownHook(new ShutdownThread());

        } catch (Exception e) {
            logger.error("Exception happen when start server", e);
        }
    }

    /**
     * blocking to wait for close.
     */
    public void waitForClose() throws InterruptedException {
        for (int i = 0; i < this.serverOptions.getWorkers(); i++) {
            bindFuture[i].channel().closeFuture().sync();
        }
    }

    public void stop() {
        logger.info("Netty server is stopping");

        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
        defaultEventExecutorGroup.shutdownGracefully();

        logger.info("Netty server stoped");
    }

    private ServerBootstrap configServer(SimpleChannelInboundHandler<CustTMessage> handler) {
        bossGroup = new EpollEventLoopGroup(this.serverOptions.getBossThreads(), new DefaultThreadFactory("ServerBossGroup"));
        workerGroup = new EpollEventLoopGroup(this.serverOptions.getWorkerThreads(), new DefaultThreadFactory("ServerWorkerGroup"));
        defaultEventExecutorGroup = new DefaultEventExecutorGroup(this.serverOptions.getEventThreads(), new DefaultThreadFactory("EventGroup"));

        ServerBootstrap b = new ServerBootstrap();
        b.group(bossGroup, workerGroup)
                .channel(EpollServerSocketChannel.class)

                .option(ChannelOption.SO_BACKLOG, this.serverOptions.getBacklog())
                .option(ChannelOption.SO_REUSEADDR, true)
                .option(EpollChannelOption.SO_REUSEPORT, true)

                .childOption(ChannelOption.TCP_NODELAY, true)

                .childOption(ChannelOption.SO_KEEPALIVE, Boolean.TRUE)
                .childOption(EpollChannelOption.TCP_KEEPIDLE, this.serverOptions.getTcpKeepIdle())
                .childOption(EpollChannelOption.TCP_KEEPCNT, 3)
                .childOption(EpollChannelOption.TCP_KEEPINTVL, this.serverOptions.getTcpKeepIdle() / 3)

                .childOption(ChannelOption.SO_SNDBUF, this.serverOptions.getSendBuf())
                .childOption(ChannelOption.SO_RCVBUF, this.serverOptions.getRecvBuf())

                .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                // 水位保护
                .childOption(ChannelOption.WRITE_BUFFER_WATER_MARK, WriteBufferWaterMark.DEFAULT);

        b.childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            public void initChannel(SocketChannel ch) throws Exception {
                ch.pipeline().addLast(
                        defaultEventExecutorGroup,
                        new LoggingHandler("com.ezcloud.debuglog", LogLevel.DEBUG),
                        new CustTMessageDecoder(),
                        new CustTMessageEncoder(),
                        new IdleStateHandler(0, 0, 60, TimeUnit.SECONDS),
                        new NettyConnectManageHandler(connectManager),
                        handler
                );
            }
        });

        return b;
    }

    class ShutdownThread extends Thread {
        public ShutdownThread() {
            super("server-shutdown");
        }

        @Override
        public void run() {
            NettyServer.this.stop();
        }
    }

    private static class NettyConnectManageHandler extends ChannelDuplexHandler {
        private NettyConnectManager connectManager;

        public NettyConnectManageHandler(NettyConnectManager connectManager) {
            this.connectManager = connectManager;
        }

        @Override
        public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
            this.connectManager.registerChannel(ctx.channel());
            super.channelRegistered(ctx);
        }

        @Override
        public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
            this.connectManager.unRegisterChannel(ctx.channel());
            super.channelUnregistered(ctx);
        }

        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            if (evt instanceof IdleStateEvent) {
                if (((IdleStateEvent) evt).state().equals(IdleState.ALL_IDLE)) {
                    this.connectManager.closeIdleChannel(ctx.channel());
                }
            }

            super.userEventTriggered(ctx, evt);
        }

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            if (msg instanceof CustTMessage && ((CustTMessage) msg).getType() == CustTType.HEARTBEAT) {
                this.connectManager.leaseRenewal(ctx.channel());
            } else {
                super.channelRead(ctx, msg);
            }
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            final String remoteAddress = RemotingUtil.parseChannelRemoteAddr(ctx.channel());
            logger.warn("NETTY SERVER PIPELINE: exceptionCaught {}, exception:", remoteAddress, cause);

            this.connectManager.closeIdleChannel(ctx.channel());
        }
    }
}
